package com.rz.spark.report

import com.rz.spark.beans.Log
import com.rz.spark.utils.RptUtil
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 广告在每个地域的投放情况统计
  *
  * 实现方式---spark core
  */
object AreaAnalyzeRptV2 {
  def main(args: Array[String]): Unit = {

    // 0 检验参数个数
    if (args.length !=2){
      println(
        """
          |com.rz.spark.report.ProCityRpt
          |参数：
          | logInputPath
          | resultOutputPath
          | """.stripMargin
      )
      sys.exit()
    }

    // 1 接受程序参数
    val Array(logInputPath, resultOutputPath) =args

    // 2 创建sparkConf-》sparkContext
    val sparkConf = new SparkConf()
    sparkConf.setAppName(s"${this.getClass.getSimpleName}")
    sparkConf.setMaster("local[*]")
    // RDD 序列化到磁盘 worker与worker之间的数据传输
    sparkConf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(sparkConf)
    val sQLContext = new SQLContext(sc)

    // 读取原始文件
    sc.textFile(logInputPath).map(_.split(",",-1)).filter(_.length>=85).map(arr=>{
      val log = Log(arr)

      val req = RptUtil.calculateReq(log.requestmode, log.processnode)
      val rtb = RptUtil.calculateRtb(log.iseffective, log.isbilling, log.isbid, log.adorderid, log.iswin, log.winprice,log.adpayment)
      val showClick = RptUtil.calculateShowClick(log.requestmode,log.iseffective)

      ((log.provincename,log.cityname),req++rtb++showClick)
    }).reduceByKey((list1,list2)=>{
      list1.zip(list2).map(t=>t._1+t._2)
    }).map(t=>t._1._1+","+t._1._2+","+t._2.mkString(",")).saveAsTextFile(resultOutputPath)




    // 读取parquet文件
    /*val parquetData: DataFrame = sQLContext.read.parquet(logInputPath)
    parquetData.map(row=>{
      // 是不是原始请求,有效请求，广告请求
      val reqMode = row.getAs[Int]("requestmode")
      val prcNode = row.getAs[Int]("processnode")
      // 参与竞价,竞价成功 List(参与竞价，竞价成功,消费，成本)
      val effective = row.getAs[Int]("iseffective")
      val bill = row.getAs[Int]("billing")
      val bid = row.getAs[Int]("isbid")
      val orderid = row.getAs[Int]("adorderid")
      val win = row.getAs[Int]("iswin")
      val winPrice = row.getAs[Double]("winprice")
      val adPayment = row.getAs[Double]("adpayment")

      val reqList= RptUtil.calculateReq(reqMode, prcNode)
      val rtbList = RptUtil.calculateRtb(effective, bill, bid,orderid, win, winPrice, adPayment)
      val showClickList = RptUtil.calculateShowClick(reqMode, effective)
      // 返回元组
      ((row.getAs[String]("provincename"),row.getAs[String]("cityname")),reqList++rtbList++showClickList)
    }).reduceByKey((list1,list2)=>{
      list1.zip(list2).map(t=>t._1+t._2)
    }).map(t=>t._1._1+","+t._1._2+","+t._2.mkString(",")).saveAsTextFile(resultOutputPath)*/


   sc.stop()

  }
}
